/*
 * Licensed to the Apache Software Foundation (ASF) under one or more
 * contributor license agreements.  See the NOTICE file distributed with
 * this work for additional information regarding copyright ownership.
 * The ASF licenses this file to You under the Apache License, Version 2.0
 * (the "License"); you may not use this file except in compliance with
 * the License.  You may obtain a copy of the License at
 *
 *     http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */

package org.apache.rocketmq.client.latency;

import org.apache.rocketmq.client.impl.producer.TopicPublishInfo;
import org.apache.rocketmq.client.log.ClientLogger;
import org.apache.rocketmq.logging.InternalLogger;
import org.apache.rocketmq.common.message.MessageQueue;

public class MQFaultStrategy {
    private final static InternalLogger log = ClientLogger.getLog();
    private final LatencyFaultTolerance<String> latencyFaultTolerance = new LatencyFaultToleranceImpl();

    private boolean sendLatencyFaultEnable = false;

    private long[] latencyMax = {50L, 100L, 550L, 1000L, 2000L, 3000L, 15000L};
    private long[] notAvailableDuration = {0L, 0L, 30000L, 60000L, 120000L, 180000L, 600000L};

    public long[] getNotAvailableDuration() {
        return notAvailableDuration;
    }

    public void setNotAvailableDuration(final long[] notAvailableDuration) {
        this.notAvailableDuration = notAvailableDuration;
    }

    public long[] getLatencyMax() {
        return latencyMax;
    }

    public void setLatencyMax(final long[] latencyMax) {
        this.latencyMax = latencyMax;
    }

    public boolean isSendLatencyFaultEnable() {
        return sendLatencyFaultEnable;
    }

    public void setSendLatencyFaultEnable(final boolean sendLatencyFaultEnable) {
        this.sendLatencyFaultEnable = sendLatencyFaultEnable;
    }

    public MessageQueue selectOneMessageQueue(final TopicPublishInfo tpInfo, final String lastBrokerName) {
        // 如果开启 sendLatencyFaultEnable  默认false
        /**
         * todo 此处是 故障延迟机制，与下面剔除机制有些不同
         * 满足 发送N条消息，当一条消息检测到broker发送消息失败时，会根据发送耗时认为当前broker在接下来的一段时间内不可用（不可用时间根据发送耗时等级来判定）
         * 此时，之后的几条消息的第一次发送时，即使选到了该Broker，也会认为该Broker不可用。
         *   如果遍历之后所有Broker都不可用，则会选择一个延迟较低的Broker进行发送，如果延迟最低的Broker
         */
        if (this.sendLatencyFaultEnable) {
            try {
                // 第一次随机一个int范围数字 之后在之前的基础上加1  Math.abs有可能为负数
                int index = tpInfo.getSendWhichQueue().getAndIncrement();
                for (int i = 0; i < tpInfo.getMessageQueueList().size(); i++) {
                    int pos = Math.abs(index++) % tpInfo.getMessageQueueList().size();
                    if (pos < 0)
                        pos = 0;
                    MessageQueue mq = tpInfo.getMessageQueueList().get(pos);
                    if (latencyFaultTolerance.isAvailable(mq.getBrokerName())) {
                        // 第一次发送时，因为mq队列Broker可用，则直接发送到该mq上
                        // 重试时，上次失败的Broker现在又可用了，仍继续发送该Broker上？
                        if (null == lastBrokerName || mq.getBrokerName().equals(lastBrokerName))
                            return mq;
                    }
                }
                // 所有队列都不可用, 根据延迟等级排序，选择延迟市场较小的一个
                final String notBestBroker = latencyFaultTolerance.pickOneAtLeast();
                int writeQueueNums = tpInfo.getQueueIdByBroker(notBestBroker);
                if (writeQueueNums > 0) {
                    final MessageQueue mq = tpInfo.selectOneMessageQueue();
                    if (notBestBroker != null) {
                        mq.setBrokerName(notBestBroker);
                        mq.setQueueId(tpInfo.getSendWhichQueue().getAndIncrement() % writeQueueNums);
                    }
                    return mq;
                } else {
                    latencyFaultTolerance.remove(notBestBroker);
                }
            } catch (Exception e) {
                log.error("Error occurred when selecting message queue", e);
            }
            // 如果延迟较小的那个Broker不可写, 则从中随机选择一个进行兜底。
            return tpInfo.selectOneMessageQueue();
        }

        /**
         * todo Broker故障剔除 机制
         *
         * 此方式虽然可以剔除在本次发送中 故障的Broker，
         * 但只针对同一个发送内的重试场景。
         * 如果 此时生产N条消息，
         * 每一条消息的第一次仍有可能打到这个故障的Broker上去。
         */
        // 默认 如果上一次发送失败，下次发送会切换broker
        return tpInfo.selectOneMessageQueue(lastBrokerName);
    }

    // todo 故障延迟机制
    /**
     * 根据消息的发送时长，匹配设置的延迟等级，进行延迟
     * 比如 本次发送消息耗时1000ms 则计算对应的延迟等级为4 则需要延迟60s
     *  此时将latencyFaultTolerance map中的该brokerName的开始服务时间设置为当前时间+60s
     *  这样在判该broker是否可用时，用 System.currentTimeMillis() > 开始服务时间 来判断！！！
     * @param brokerName
     * @param currentLatency
     * @param isolation
     */
    public void updateFaultItem(final String brokerName, final long currentLatency, boolean isolation) {
        if (this.sendLatencyFaultEnable) {
            long duration = computeNotAvailableDuration(isolation ? 30000 : currentLatency);
            this.latencyFaultTolerance.updateFaultItem(brokerName, currentLatency, duration);

        }
    }

    private long computeNotAvailableDuration(final long currentLatency) {
        for (int i = latencyMax.length - 1; i >= 0; i--) {
            if (currentLatency >= latencyMax[i])
                return this.notAvailableDuration[i];
        }

        return 0;
    }
}
